package com.vodafone.mapreduce.listener;

import java.util.ArrayList;
import java.util.List;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.ObjectMessage;

import org.apache.log4j.Logger;

import com.vodafone.mapreduce.devide.WordFrequencyDivideTest;
import com.vodafone.mapreduce.dto.function.MapInput;
import com.vodafone.mapreduce.processor.MapInputDtoProcessor;

public class MapInputDtoListener<K, V> extends Thread implements MessageListener {

    private final Logger logger;
    private final MapInputDtoProcessor processor;
    private final List<MapInput<K, V>> received = new ArrayList<MapInput<K, V>>();

    public MapInputDtoListener(Logger logger, MapInputDtoProcessor processor) {
        this.logger = logger;
        this.processor = processor;
    }

    public List<MapInput<K, V>> getReceived() {
        return received;
    }

    @Override
    public void onMessage(Message message) {
        WordFrequencyDivideTest.LOG.info("Received message: " + message.toString());
        MapInput<K, V> c = null;

        if (message instanceof ObjectMessage) {
            if (logger.isDebugEnabled()) {
                logger.debug("Message is an ObjectMessage.");
            }
            final ObjectMessage objectMessage = (ObjectMessage) message;
            try {
                final Object o = objectMessage.getObject();
                if (o instanceof MapInput<?, ?>) {
                    c = (MapInput<K, V>) o;
                    logger.info("Received dto: key = " + c.getKey().toString() + ", value = " + c.getValue().toString());
                }
                // processor.process(c);
                received.add(c);
            } catch (final JMSException e) {
                if (e.getCause() instanceof ClassNotFoundException) {
                    this.logger.warn(this.getClass().getSimpleName() + " invalid class, will ACK it, no sense in reprocessing.",
                            e.getCause());
                    return;
                }
            }
        } else {
            logger.warn("We're expecting an object message!");
        }

    }

}